package cas.ihep.fs.alluxio;

import alluxio.AlluxioURI;
import alluxio.client.file.FileSystem;
import alluxio.client.file.FileSystemContext;
import alluxio.client.file.FileSystemUtils;
import alluxio.client.file.options.CreateDirectoryOptions;
import alluxio.client.file.options.CreateFileOptions;
import alluxio.exception.AlluxioException;
import alluxio.security.authorization.Mode;
import cas.ihep.fs.UnifiedAbstractFileSystem;
import cas.ihep.fs.UnifiedFile;
import cas.ihep.fs.UnifiedFileSystem;

import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.lang.reflect.Field;
import java.net.URI;

public class AlluxioFileSystem implements UnifiedAbstractFileSystem {

    FileSystemContext fileSystemContext;
    FileSystem fileSystem;
    AlluxioInvoker alluxioInvoker;
    private boolean isClosed;
    public AlluxioFileSystem(UnifiedFileSystem.Configuration uConf)throws Exception{
        fileSystemContext=FileSystemContext.create();
        fileSystem=FileSystem.Factory.get(fileSystemContext);
        alluxioInvoker=new AlluxioInvoker(fileSystemContext);
        isClosed=false;
    }

    public FileSystem getFileSystem() {
        return fileSystem;
    }

    @Override
    public UnifiedFile open(URI path, UnifiedFileSystem.OpenOptions options) throws IOException {
        return new AlluxioFile.AlluxioFileIn(this,newURI(path));
    }

    @Override
    public UnifiedFile create(URI path, UnifiedFileSystem.CreateOptions options) throws IOException {
        return new AlluxioFile.AlluxioFileOut(this,newURI(path),options);
    }

    @Override
    public InputStream openStream(URI path, UnifiedFileSystem.OpenOptions options) throws IOException {
        try {
            return fileSystem.openFile(newURI(path.normalize()));
        }catch (AlluxioException e){
            throw new IOException(e.getMessage());
        }
    }

    @Override
    public OutputStream createStream(URI path, UnifiedFileSystem.CreateOptions options) throws IOException {
        try{
            CreateFileOptions cfopt=CreateFileOptions.defaults();
            cfopt.setBlockSizeBytes(options.blockSize());
            cfopt.setRecursive(options.createParent());
            cfopt.setMode(new Mode(options.permission()));
            return fileSystem.createFile(newURI(path.normalize()),cfopt);
        }catch (AlluxioException e){
            throw new IOException(e.getMessage());
        }
    }

    @Override
    public void persist(URI path) throws IOException {
        FileSystemUtils.persistFile(fileSystem,newURI(path));
    }

    @Override
    public void persist(UnifiedFile file) throws IOException {
        if(file instanceof AlluxioFile){
            AlluxioFile afile=(AlluxioFile)file;
            FileSystemUtils.persistFile(fileSystem,afile.alluxioUri);
            return;
        }
        throw new IOException("Unsupported operation");
    }

    @Override
    public void free(URI path) throws IOException {
        try {
            fileSystem.free(newURI(path));
        } catch (AlluxioException e) {
            throw new IOException(e);
        }
    }

    @Override
    public void free(UnifiedFile file) throws IOException {
        if(file instanceof AlluxioFile){
            AlluxioFile afile=(AlluxioFile)file;
            try {
                fileSystem.free(afile.alluxioUri);
                return;
            }catch (AlluxioException e){
                throw new IOException(e);
            }
        }
        throw new IOException("Unsupported operation");
    }

    @Override
    public void delete(URI path) throws IOException {
        try {
            fileSystem.delete(newURI(path));
        } catch (AlluxioException e) {
            throw new IOException(e);
        }
    }

    @Override
    public void delete(UnifiedFile file) throws IOException {
        if(file instanceof AlluxioFile){
            AlluxioFile afile=(AlluxioFile)file;
            try{
                fileSystem.delete(afile.alluxioUri);
                return;
            }catch (AlluxioException e){
                throw new IOException(e);
            }
        }
        throw new IOException("Unsupported operation");
    }

    @Override
    public void mkdir(URI path,UnifiedFileSystem.CreateOptions options) throws IOException {
        try {
            CreateDirectoryOptions opts=CreateDirectoryOptions.defaults();
            short sh=options.permission();
            byte low= (byte) (sh&7);
            byte mid= (byte) ((sh>>3)&7);
            byte high= (byte) ((sh>>6)&7);
            if((high & 6)!=0){
                high|=1;
            }
            if((mid&6)!=0){
                mid|=1;
            }
            if((low&6)!=0){
                low|=1;
            }
            sh= (short) ((sh&0xfe00)|(high<<6)|(mid<<3)|low);
            opts.setMode(new Mode(sh));
            opts.setRecursive(options.createParent());
            opts.setAllowExists(options.overwrite());
            fileSystem.createDirectory(newURI(path),opts);
        } catch (AlluxioException e) {
            throw new IOException(e);
        }
    }

    @Override
    public void close() throws IOException {
        if(!isClosed) {
            isClosed=true;
//            if (fileSystem instanceof Closeable) {
//                ((Closeable) fileSystem).close();
//            }
            try {
                if (fileSystem instanceof Closeable) {
                    ((Closeable) fileSystem).close();
                }
            } finally {
                fileSystemContext.close();
            }
        }
    }

    private static final Class<AlluxioURI> alluxioURIClass;
    private static final Field mUriField;

    static {
        if(System.getProperty("alluxio.site.conf.dir")==null){
            String alluxio_home=System.getenv("ALLUXIO_HOME");
            if(alluxio_home!=null){
                alluxio_home=alluxio_home+"/conf";
                System.setProperty("alluxio.site.conf.dir",alluxio_home);
                System.out.println("Setting alluxio.site.conf.dir="+alluxio_home);
            }
        }
        try{
            //ClassPool pool=ClassPool.getDefault();
            //CtClass ctAlluxioURIClass=pool.get("alluxio.AlluxioURI");
            //ctAlluxioURIClass.addConstructor(CtNewConstructor.defaultConstructor(ctAlluxioURIClass));
            //noinspection unchecked
            //alluxioURIClass= (Class<AlluxioURI>) ctAlluxioURIClass.toClass();
            alluxioURIClass=AlluxioURI.class;
            //noinspection JavaReflectionMemberAccess
            mUriField=alluxioURIClass.getDeclaredField("mUri");
            mUriField.setAccessible(true);
        }catch (RuntimeException e){
            throw e;
        }catch (Exception e){
            throw new RuntimeException(e);
        }
    }

    private static AlluxioURI newURI(java.net.URI baseUri){
        StdURI uri=new StdURI(baseUri);
        try {
            AlluxioURI ret=new AlluxioURI("");
            mUriField.set(ret,uri);
            return ret;
        } catch (RuntimeException e) {
            throw e;
        } catch (Exception e){
            throw new RuntimeException(e);
        }
    }

    static URI baseURI(AlluxioURI apUri){
        try {
            alluxio.URI aUri = (alluxio.URI) mUriField.get(apUri);
            return aUri.getBaseURI();
        }catch (RuntimeException e){
            throw e;
        }catch (Exception e){
            throw new RuntimeException(e);
        }
    }
}
